package edu.cornell.cs.lsi.mapreduce.pass1;

import static edu.cornell.cs.lsi.mapreduce.Constants.COLUMN_COUNT;
import static edu.cornell.cs.lsi.mapreduce.Constants.G;
import static edu.cornell.cs.lsi.mapreduce.Constants.Lmax_SQUARED;

import java.io.IOException;
import java.io.PrintWriter;
import java.net.URI;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import edu.cornell.cs.lsi.mapreduce.Constants;

public class UnionMaker {

    private static enum Pass1Counter {
        VERTICES, EDGES
    }

    public static class Map extends Mapper<LongWritable, Text, Text, Text> {

        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            String[] split = value.toString().split(" ");

            double x = Double.parseDouble(split[0]);
            double y = Double.parseDouble(split[1]);
            double weight = Double.parseDouble(split[2]);

            if (Constants.MIN_WEIGHT <= weight && weight <= Constants.MAX_WEIGHT) {
                double colNumDouble = x / G;
                int colNumInt = (int) colNumDouble;
                double offset = x - ((double) colNumInt) * G;
                Text pointText = new Text(x + " " + y);
                if (offset >= Constants.BORDERLESS_COL_WIDTH && colNumInt < COLUMN_COUNT - 1) {
                    // it is in the left boundary of the next column or on the current right boundary
                    context.write(new Text(String.valueOf(colNumInt + 1)), pointText);
                } else if (colNumDouble == Math.ceil(colNumDouble) && colNumInt > 0) {
                    // right boundary of the previous column or on the current left boundary
                    context.write(new Text(String.valueOf(colNumInt - 1)), pointText);
                }
                context.write(new Text(String.valueOf(colNumInt)), pointText);
                context.getCounter(Pass1Counter.VERTICES).increment(1);
            }

        }
    }

    public static class Reduce extends Reducer<Text, Text, Text, Text> {

        @Override
        public void reduce(Text key, Iterable<Text> value, Context context) throws IOException, InterruptedException {
            int colNum = Integer.parseInt(key.toString());
            int seekIndex;
            int remIndex;
            int numEdges = 0;
            DoublePoint curPoint;
            SortedSet<DoublePoint> pointsInSpan;
            TreeSet<DoublePoint> tree;
            DoublePoint minPoint;
            DoublePoint maxPoint;
            double leftColBoundary = ((double) (colNum)) * G;
            double rightColBoundary = leftColBoundary + Constants.BORDERLESS_COL_WIDTH;

            Comparator<DoublePoint> yComp = new Comparator<DoublePoint>() {
                @Override
                public int compare(DoublePoint arg0, DoublePoint arg1) {
                    if (arg0.getY() < arg1.getY())
                        return -1;
                    if (arg0.getY() > arg1.getY())
                        return 1;
                    if (arg0.getX() < arg1.getX())
                        return -1;
                    if (arg0.getX() > arg1.getX())
                        return 1;
                    return 0;
                }
            };

            Comparator<DoublePoint> xComp = new Comparator<DoublePoint>() {
                @Override
                public int compare(DoublePoint arg0, DoublePoint arg1) {
                    if (arg0.getX() < arg1.getX())
                        return -1;
                    if (arg0.getX() > arg1.getX())
                        return 1;
                    if (arg0.getY() < arg1.getY())
                        return -1;
                    if (arg0.getY() > arg1.getY())
                        return 1;
                    return 0;
                }
            };

            tree = new TreeSet<DoublePoint>(yComp);

            List<DoublePoint> allPoints = new ArrayList<DoublePoint>();
            List<DoublePoint> edgePoints = new ArrayList<DoublePoint>();

            for (Text t : value) {
                allPoints.add(DoublePoint.parseFromString(t.toString()));
            }
            Collections.sort(allPoints, xComp);

            remIndex = 0;
            for (seekIndex = 0; seekIndex < allPoints.size(); seekIndex++) {
                curPoint = allPoints.get(seekIndex);
                minPoint = new DoublePoint(-1f, curPoint.getY() - Constants.Lmax);
                maxPoint = new DoublePoint(2f, curPoint.getY() + Constants.Lmax);
                pointsInSpan = tree.subSet(minPoint, maxPoint);

                if (curPoint.getX() <= leftColBoundary || curPoint.getX() >= rightColBoundary) {
                    edgePoints.add(curPoint);
                }

                for (DoublePoint pointInSpan : pointsInSpan) {
                    if (pointInSpan.withinRange(curPoint, Lmax_SQUARED)) {
                        if (curPoint.getX() >= leftColBoundary && curPoint.getX() < leftColBoundary + G) {
                            numEdges++;
                        }
                        pointInSpan.union(curPoint);
                    }
                }

                if (seekIndex + 1 < allPoints.size()) {
                    while (remIndex < seekIndex && allPoints.get(remIndex).getX() < curPoint.getX() - Constants.Lmax) {
                        tree.remove(allPoints.get(remIndex));
                        remIndex++;
                    }
                }
                tree.add(curPoint);
            }

            context.getCounter(Pass1Counter.EDGES).increment(numEdges);

            for (DoublePoint f : edgePoints) {
                context.write(new Text(f.toString()), new Text(f.find().toString()));
            }
        }
    }

    public static void main(String[] args) throws Exception {

        Job job = new Job();
        job.setJobName("unionmaker");

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        job.setJarByClass(UnionMaker.class);
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);

        job.setNumReduceTasks(COLUMN_COUNT);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        job.waitForCompletion(false);
        long numVertices = job.getCounters().findCounter(Pass1Counter.VERTICES).getValue();
        long numEdges = job.getCounters().findCounter(Pass1Counter.EDGES).getValue();

        Path statsFile = new Path(args[2]);
        FileSystem fs = FileSystem.get(URI.create(args[2]), job.getConfiguration());
        PrintWriter statsFileOut = new PrintWriter(fs.create(statsFile, true));
        statsFileOut.println("Number of Vertices: " + numVertices);
        statsFileOut.println("Number of Edges: " + numEdges);
        statsFileOut.flush();
        statsFileOut.close();
    }
}
